package com.heima.schedule.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojos.Taskinfo;
import com.heima.model.schedule.pojos.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Service//实现类注解
@Transactional//因为涉及到多张表的更改操作，所以需要开启事务，保证数据的一致性
@Slf4j//开启日志
public class TaskServiceImpl implements TaskService {

	/**
	 * 添加延迟任务
	 *
	 * @param task
	 * @return
	 */
	@Override
	public long addTask(Task task) {
		//1.添加任务到数据库中
		boolean success = addTaskToDb(task);

		//如果任务添加到数据库中成功，则调用该方法，执行添加任务到redis中
		if (success) {
			//2.当任务添加到数据库成功，才添加任务到redis
			addTaskToCache(task);
		}
		//3.返回任务id
		return task.getTaskId();
	}

	@Autowired
	private CacheService cacheService;

	/**
	 * 把任务添加到redis中
	 *
	 * @param task
	 */
	private void addTaskToCache(Task task) {
		//创建key，通过任务的类型和优先级来确定唯一的Redis键
		String key = task.getTaskType() + "_" + task.getPriority();

		//获取5分钟之后的时间  毫秒值
		//创建一个日期对象，获取当前时间的Calendar对象，并设置5分钟之后的时间
		Calendar calendar = Calendar.getInstance();
		//add方法的第一个参数指定了要修改的时间字段，这里是Calendar.MINUTE，表示分钟
		//第二个参数是要添加的时间量，这里是 5，表示增加 5 分钟
		calendar.add(Calendar.MINUTE, 5);
		//获取时间戳
		long nextScheduleTime = calendar.getTimeInMillis();

		//2.1 如果任务的执行时间小于等于当前时间，存入list
		//如果任务的执行时间小于等于当前时间，这意味着任务已经过期或应该立即执行，所以需要立即执行
		if (task.getExecuteTime() <= System.currentTimeMillis()) {
			//立即执行，存入list
			cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
		} else if (task.getExecuteTime() <= nextScheduleTime) {
			//2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
			cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
		}
	}

	@Autowired
	private TaskinfoMapper taskinfoMapper;

	@Autowired
	private TaskinfoLogsMapper taskinfoLogsMapper;

	/**
	 * 添加任务到数据库中
	 *
	 * @param task
	 * @return
	 */
	private boolean addTaskToDb(Task task) {
		//定义一个布尔类型的变量flag，用于标记任务是否成功添加到数据库中，初始值为false
		boolean flag = false;
		try {
			//保存任务表
			//创建一个新的Taskinfo对象，用于存储任务信息
			Taskinfo taskinfo = new Taskinfo();
			BeanUtils.copyProperties(task, taskinfo);
			//因为task和taskinfo中executeTime的数据类型不一样，所以需要额外处理
			//taskinfo中executeTime的数据类型是Date，而task中executeTime的数据类型是long，所以需要转换一下
			//new Date(...)：创建一个新的Date对象，该对象表示task.getExecuteTime()方法返回的时间值
			taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
			//保存任务信息
			taskinfoMapper.insert(taskinfo);

			//设置taskID，将新插入的任务记录生成的ID设置到原始的Task对象中
			task.setTaskId(taskinfo.getTaskId());

			//保存任务日志数据
			TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
			BeanUtils.copyProperties(taskinfo, taskinfoLogs);
			taskinfoLogs.setVersion(1);//初始化版本号为1，用于跟踪任务的执行情况
			//ScheduleConstants是常量类，里面定义了一些常量，如任务状态、任务类型等
			taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);//初始化状态为0，即初始化状态
			taskinfoLogsMapper.insert(taskinfoLogs);//保存任务日志数据
			//代码执行到这里，表示以及向任务表和任务日志表插入了数据，表示任务添加成功
			flag = true;
		} catch (Exception e) {
			//如果出现异常，打印异常信息
			e.printStackTrace();
		}
		//返回flag，表示任务是否成功添加到数据库中
		return flag;
	}

	/**
	 * 取消任务
	 *
	 * @param taskId
	 * @return
	 */
	@Override
	public boolean cancelTask(long taskId) {

		//定义一个布尔类型的变量 flag，用于标记任务是否成功取消，初始值为 false
		boolean flag = false;

		//删除任务，更新日志
		Task task = updateDb(taskId, ScheduleConstants.CANCELLED);
		//删除redis的数据
		if (task != null) {//task有数据，表示数据库中的数据已经更新成功
			removeTaskFromCache(task);
			//代码执行到这里，表示redis中的任务取消成功
			flag = true;
		}
		//代码执行到这里，表示任务取消成功，返回flag
		return flag;
	}

	/**
	 * 删除redis中的任务对象
	 *
	 * @param task
	 */
	private void removeTaskFromCache(Task task) {
		//使用任务的类型 taskType 和优先级 priority 来创建一个唯一的缓存键
		String key = task.getTaskType() + "_" + task.getPriority();
		//这段代码检查任务的执行时间 executeTime 是否小于或等于当前时间（System.currentTimeMillis()）
		if (task.getExecuteTime() <= System.currentTimeMillis()) {
			//如果是，则表示该任务应该已经执行或者已经过期，因此它会从 Redis 的列表（ScheduleConstants.TOPIC + key）中移除
			//JSON.toJSONString(task) 将任务对象转换为 JSON 字符串，以便从 Redis 中删除
			cacheService.lRemove(ScheduleConstants.TOPIC + key, 0, JSON.toJSONString(task));
		} else {
			//如果任务的执行时间大于当前时间，则说明该任务还没有到执行的时间点
			//因此，它会从 Redis 的有序集合（ScheduleConstants.FUTURE + key）中移除
			cacheService.zRemove(ScheduleConstants.FUTURE + key, JSON.toJSONString(task));
		}
	}

	/**
	 * 删除任务，更新任务日志
	 *
	 * @param taskId
	 * @param status
	 * @return
	 */
	private Task updateDb(long taskId, int status) {
		//创建一个全局任务对象，task
		Task task = null;
		try {
			//删除任务,同时需要通过任务id去任务信息表中删除指定数据
			taskinfoMapper.deleteById(taskId);

			//更新任务日志,根据任务id查询任务日志表中的数据
			TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
			//更新该条数据中的任务状态，status是方法传递进来的参数，即取消状态
			taskinfoLogs.setStatus(status);
			//根据任务日志对象更新任务日志表中的该条数据
			taskinfoLogsMapper.updateById(taskinfoLogs);

			task = new Task();
			//属性拷贝
			BeanUtils.copyProperties(taskinfoLogs, task);
			//getTime()方法是Date类的一个方法，用于返回Date对象所表示的时间点的毫秒值
			//即从1970年1月1日00:00:00 GMT到该时间点的毫秒数
			//这里的逻辑是将 taskinfoLogs 中的 executeTime 转换为 long 类型，并赋值给 task 的 executeTime 属性
			task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
		} catch (Exception e) {
			log.error("task cancel exception taskid={}", taskId);
		}
		return task;
	}

	/**
	 * 按照类型和优先级拉取任务，即消费任务
	 * type（任务类型）和 priority（任务优先级）
	 *
	 * @return
	 */
	@Override
	public Task poll(int type, int priority) {
		//声明一个 Task 类型的变量 task，初始值为 null，用于存储从 Redis 中拉取的任务
		Task task = null;
		try {
			//根据传入的任务类型和优先级，构建一个唯一的 Redis
			//因为我们存入任务的时候是以“任务类型和优先级”为key的，所以取的时候也是以“任务类型和优先级”为key的
			String key = type + "_" + priority;
			//从 Redis 中的列表（ScheduleConstants.TOPIC + key）中弹出一个任务的 JSON 字符串
			//lRightPop 方法会从列表的右侧（尾部）弹出一个元素，并将其从列表中移除
			String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
			//task_json 不为空，则继续执行后续操作
			//使用 StringUtils.isNotBlank 方法检查拉取到的 JSON 字符串是否为空
			if (StringUtils.isNotBlank(task_json)) {
				//使用 JSON.parseObject 方法将 JSON 字符串解析为 Task 对象
				task = JSON.parseObject(task_json, Task.class);
				//更新数据库信息，将状态改为已经执行状态
				updateDb(task.getTaskId(), ScheduleConstants.EXECUTED);
			}
		} catch (Exception e) {
			e.printStackTrace();
			log.error("poll task exception");
		}
		//返回从 Redis 中拉取并执行的任务对象
		return task;
	}

	/**
	 * 未来数据定时刷新
	 */
	//主要功能是定期检查 Redis 中存储的未来任务（即那些需要在未来的某个时间点执行的任务），并将其移动到需要立即执行的任务队列中
	//这样可以确保任务在需要执行的时间点能够被及时获取并处理
	@Scheduled(cron = "0 */1 * * * ?")//表示每分钟执行一次，在每分钟的第0秒执行一次
	public void refresh() {

		//tryLock 方法是 CacheService 提供的一个用于尝试获取分布式锁的方法
		//"FUTURE_TASK_SYNC"：这是一个锁的名称;1000 * 30：这是锁的超时时间，以毫秒为单位
		//token 是一个唯一的令牌，如果成功获取锁，则 token 不为空，否则为空
		//如果获取锁成功，则执行定时任务，否则跳过该定时任务
		String token = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);
		//如果成功获取锁，tryLock 方法会返回一个唯一的 token（令牌），对token进行判断
		if (StringUtils.isNotBlank(token)) {
			//打印当前时间戳（以秒为单位）和一条消息，表示定时任务已经执行，用来测试定时任务是否正常执行
			System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
			// 获取所有未来数据集合的key值，使用 cacheService.scan 方法来获取所有以 ScheduleConstants.FUTURE 开头的 Redis 键
			Set<String> futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
			//遍历 futureKeys 集合中的每一个键
			for (String futureKey : futureKeys) { // future_250_250-->{[future_][250_250]}
				//对于每一个 futureKey，通过 split 方法去除 ScheduleConstants.FUTURE 部分
				//然后在前面加上 ScheduleConstants.TOPIC，从而构建出对应的 topicKey
				//ScheduleConstants.TOPIC 通常为 "topic_"，用于表示当前需要执行的任务的主题
				// future_250_250-->{[future_][250_250]}---->下标为1即为250_250
				//最后组成"topic_"+250_250即topic_250_250格式的字符串
				String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
				//获取该组key下当前需要消费的任务数据
				Set<String> tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
				//同步数据
				if (!tasks.isEmpty()) {
					//将这些任务数据添加到消费者队列中,refreshWithPipeline 方法使用了 Redis 的管道技术来提高性能
					cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
					//打印一条消息表示成功的将任务数据同步到消费者队列
					System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
				}
			}
		}
	}

	/**
	 * 数据库任务定时同步到redis
	 */
	@Scheduled(cron = "0 */5 * * * ?")//表示每5分钟执行一次，在每分钟的第0秒执行一次
	@PostConstruct//当前微服务宕机后，重启后，会自动执行该方法恢复数据
	public void reloadData() {
		//清空缓存
		clearCache();
		log.info("数据库数据同步到缓存");
		//获取当前时间的Calendar对象，并设置5分钟之后的时间
		Calendar calendar = Calendar.getInstance();
		calendar.add(Calendar.MINUTE, 5);

		//查看小于未来5分钟的所有任务
		List<Taskinfo> allTasks = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime,calendar.getTime()));
		//遍历任务列表，将任务添加到redis中
		if(allTasks != null && allTasks.size() > 0){
			for (Taskinfo taskinfo : allTasks) {
				Task task = new Task();
				//属性拷贝
				BeanUtils.copyProperties(taskinfo,task);
				//给task对象设置executeTime属性，因为taskinfo中executeTime的数据类型是Date，而task中executeTime的数据类型是long，所以需要转换一下
				task.setExecuteTime(taskinfo.getExecuteTime().getTime());
				//添加到redis缓存中
				addTaskToCache(task);
			}
		}
		log.info("数据库数据同步到redis缓存完成");
	}

	//自定义方法clearCache，用于清空redis的缓存，这个方法在数据库向redis同步数据前调用
	//在数据库数据同步到redis缓存前清空redis缓存，这样可以保证redis缓存中只有最新的数据
	private void clearCache(){
		// 删除缓存中未来数据集合和当前消费者队列的所有key
		Set<String> futurekeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_
		Set<String> topickeys = cacheService.scan(ScheduleConstants.TOPIC + "*");// topic_
		// 删除缓存中未来数据集合和当前消费者队列的所有key
		cacheService.delete(topickeys);
		cacheService.delete(futurekeys);
	}
}